package bingo.rpc.framework.zookeeper;

import bingo.rpc.framework.helper.IPHelper;
import bingo.rpc.framework.helper.PropertyConfigHelper;
import bingo.rpc.framework.model.InvokerService;
import bingo.rpc.framework.model.ProviderService;
import com.alibaba.fastjson.JSON;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.SerializableSerializer;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * 注册中心
 *
 * @author hujinsheng
 * @project bingo
 * @create 2017-12-11 14:34
 **/
public class RegisterCenter implements IRegisterCenter4Provider ,IRegisterCenter4Consumer {
    Logger logger= LoggerFactory.getLogger(RegisterCenter.class);

    private static RegisterCenter registerCenter=new RegisterCenter();
    //服务提供者列表,Key:服务提供者接口  value:服务提供者服务方法列表
    private static final Map<String,List<ProviderService>> providerServiceMap=Maps.newConcurrentMap();
    private static final Map<String,List<ProviderService>> serviceMetaDataMap4Consumer=Maps.newConcurrentMap();
    private static volatile ZkClient zkClient = null;
    private static String ZK_SERVICE = PropertyConfigHelper.getZkService();
    private static int ZK_SESSION_TIME_OUT = PropertyConfigHelper.getZkConnectionTimeout();
    private static int ZK_CONNECTION_TIME_OUT = PropertyConfigHelper.getZkConnectionTimeout();
    private static String ROOT_PATH = "/bingo";
    public static String PROVIDER_TYPE = "provider";
    public static String INVOKER_TYPE = "consumer";
    @Override
    public void registerProvider(List<ProviderService> serviceMetaData) {

        if(CollectionUtils.isEmpty(serviceMetaData)){
            return;
        }
        synchronized (RegisterCenter.class){
            for(ProviderService provider:serviceMetaData){
                String interfaceClass=provider.getInterfaceClass().getName();
                List<ProviderService> providers = providerServiceMap.get(interfaceClass);
                if(providers==null){
                    providers=new  ArrayList();
                }
                providers.add(provider);
                providerServiceMap.put(interfaceClass,providers);
            }

            if (zkClient == null) {
                zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
            }
            String APP_KEY=serviceMetaData.get(0).getAppKey();
            String ZK_PATH=ROOT_PATH+"/"+APP_KEY;
            boolean exist=zkClient.exists(ZK_PATH);
            if(!exist){
                zkClient.createPersistent(ZK_PATH,true);
            }

           for(Map.Entry<String,List<ProviderService>> entry:providerServiceMap.entrySet()){
               //服务分组
               String groupName = entry.getValue().get(0).getGroupName();
               //创建服务提供者
               String serviceNode = entry.getKey();
               String servicePath = ZK_PATH + "/" + groupName + "/" + serviceNode + "/" + PROVIDER_TYPE;
               exist = zkClient.exists(servicePath);
               if (!exist) {
                   zkClient.createPersistent(servicePath, true);
               }


               //创建当前服务器节点
               int serverPort = entry.getValue().get(0).getServerPort();//服务端口
               int weight = entry.getValue().get(0).getWeight();//服务权重
               int workerThreads = entry.getValue().get(0).getWorkerThreads();//服务工作线程
               String localIp = IPHelper.getLocalIp();
               String currentServiceIpNode=servicePath+"/"+localIp+"|"+serverPort+"|"+weight+"|"+workerThreads+"|"+groupName;
               exist=zkClient.exists(currentServiceIpNode);
               if(!exist){
                   //创建临时节点
                   zkClient.createEphemeral(currentServiceIpNode);
               }
               //监听注册服务的变化，同事更新数据到本地缓存
               zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
                   @Override
                   public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                        if(null==currentChilds){
                            currentChilds=new ArrayList<String>();
                        }
                       List<String> activityServiceIpList= Lists.newArrayList(Lists.transform(currentChilds, new Function<String, String>() {
                           @Override
                           public String apply(String input) {
                               return StringUtils.split(input,"|")[0];
                           }
                       }));
                       refreshActivityService(activityServiceIpList);
                   }
               });
            }
        }
    }
    @Override
    public Map<String, List<ProviderService>> getProvideServiceMap() {
        return providerServiceMap;
    }
    @Override
    public void initProvidermap(String remoteAppKey, String groupName) {
        if(MapUtils.isEmpty(serviceMetaDataMap4Consumer)){
            serviceMetaDataMap4Consumer.putAll(fetchOrUpdateServiceMetaData(remoteAppKey,groupName));
        }

    }
    private Map<String,List<ProviderService>> fetchOrUpdateServiceMetaData(String remoteAppKey, String groupName) {
        final Map<String,List<ProviderService>> providerServiceMap=Maps.newConcurrentMap();
        synchronized (RegisterCenter.class){
            if(zkClient==null){
                zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
            }
        }
        String providePath=ROOT_PATH+"/"+remoteAppKey+"/"+groupName;
        List<String> providerServices=zkClient.getChildren(providePath);
        for(String serviceName:providerServices){
            String servicePath=providePath+"/"+serviceName+"/"+PROVIDER_TYPE;
            List<String> ipPathList=zkClient.getChildren(servicePath);
            for(String ipPath:ipPathList){
                String serverIp=StringUtils.split(ipPath,"|")[0];
                String serverPort=StringUtils.split(ipPath,"|")[1];
                int weight=Integer.parseInt(StringUtils.split(ipPath,"|")[2]);
                int workerThreads=Integer.parseInt(StringUtils.split(ipPath,"|")[3]);
                String group=StringUtils.split(ipPath,"|")[4];
                List<ProviderService> providerServiceList=providerServiceMap.get(serviceName);
                if(CollectionUtils.isEmpty(providerServiceList)){
                    providerServiceList=Lists.newArrayList();
                }
                ProviderService providerService=new ProviderService();
                try{
                    providerService.setInterfaceClass(serviceName.getClass());
                }catch (Exception e){
                    throw new RuntimeException(e);
                }
                providerService.setServerIp(serverIp);
                providerService.setServerPort(Integer.parseInt(serverPort));
                providerService.setGroupName(groupName);
                providerService.setWeight(weight);
                providerService.setWorkerThreads(workerThreads);
                providerServiceList.add(providerService);
                providerServiceMap.put(serviceName,providerServiceList);
            }
            //监听注册服务的变化,同时更新数据到本地缓存
            zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
                @Override
                public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
                    if(null==currentChilds){
                        currentChilds=Lists.newArrayList();
                    }
                    Lists.newArrayList(Lists.transform(currentChilds, new Function<String, String>() {

                        @Override
                        public String apply(String input) {
                            return StringUtils.split(input,"|")[0];
                        }
                    }));
                    refreshServiceMetaDataMap(currentChilds);
                }
            });
        }
        return null;
    }
    /**
     * 利用zk自动刷新当前存活的服务提供者列表数据
     * @param serviceIpList
     */
    private void refreshActivityService(List<String> serviceIpList) {
        if (serviceIpList == null) {
            serviceIpList = Lists.newArrayList();
        }
        Map<String,List<ProviderService>> currentServiceMetaDataMap= Maps.newConcurrentMap();
        for(Map.Entry<String,List<ProviderService>> entry:providerServiceMap.entrySet()){
            String key=entry.getKey();
            List<ProviderService> providerServices=entry.getValue();
            List<ProviderService> serviceMataDataModelList = currentServiceMetaDataMap.get(key);
            if(CollectionUtils.isEmpty(serviceMataDataModelList)){
                serviceMataDataModelList=Lists.newArrayList();
            }
            for(ProviderService serviceMataData:providerServices){
                if(serviceIpList.contains(serviceMataData.getServerIp())){
                    serviceMataDataModelList.add(serviceMataData);
                }
            }
            currentServiceMetaDataMap.put(key,serviceMataDataModelList);
        }
        providerServiceMap.clear();
        logger.info("currentServiceMetaDataMap,"+ JSON.toJSONString(currentServiceMetaDataMap));
        providerServiceMap.putAll(currentServiceMetaDataMap);
    }
    /**
     * 刷新
     * @param serviceIpList
     */
    private void refreshServiceMetaDataMap(List<String> serviceIpList) {

        if(serviceIpList==null){
            serviceIpList=Lists.newArrayList();
        }
        Map<String,List<ProviderService>> currentServiceMetaDataMap=Maps.newConcurrentMap();
        for(Map.Entry<String,List<ProviderService>> entry:serviceMetaDataMap4Consumer.entrySet()){
            String key=entry.getKey();
            List<ProviderService> serviceList=entry.getValue();
            List<ProviderService> serviceMetaModelList=currentServiceMetaDataMap.get(key);
            if(CollectionUtils.isEmpty(serviceMetaModelList)){
                serviceMetaModelList=Lists.newArrayList();
            }
            for(ProviderService providerService:serviceList){
                if(serviceIpList.contains(providerService.getServerIp())){
                    serviceMetaModelList.add(providerService);
                }
            }
            currentServiceMetaDataMap.put(key,serviceMetaModelList);
        }
        serviceMetaDataMap4Consumer.clear();
        serviceMetaDataMap4Consumer.putAll(currentServiceMetaDataMap);
    }
    @Override
    public Map<String, List<ProviderService>> getServiceMetaDataMap4Consumer() {
        return serviceMetaDataMap4Consumer;
    }

    @Override
    public void registerInvoker(InvokerService invoker) {
        if(null==invoker){
            return;
        }
        synchronized (RegisterCenter.class){
            if(zkClient==null){
                zkClient = new ZkClient(ZK_SERVICE, ZK_SESSION_TIME_OUT, ZK_CONNECTION_TIME_OUT, new SerializableSerializer());
            }
            //创建zk命名空间
            boolean exist=zkClient.exists(ROOT_PATH);
            if(!exist){
                zkClient.createPersistent(ROOT_PATH,true);
            }
            //创建消费者节点
            String remoteAppKey=invoker.getRemoteAppKey();
            String groupName=invoker.getGroupName();
            String serviceNode=invoker.getInterfaceClass().getName();
            String servicePath=ROOT_PATH+"/"+remoteAppKey+"/"+groupName+"/"+serviceNode+"/"+INVOKER_TYPE;
            exist=zkClient.exists(servicePath);
            if(!exist){
                zkClient.createPersistent(servicePath,true);
            }
            //创建当前服务器节点
            String localIp=IPHelper.getLocalIp();
            String currentServiceIpNode=servicePath+"/"+localIp;
            exist=zkClient.exists(currentServiceIpNode);
            if(!exist){
                zkClient.createEphemeral(currentServiceIpNode);
            }

        }

    }
}
